package com.atguigu.dataStreamApi.transformation;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Objects;

public class Flink08_UnionConnectStream {
      public static void main(String[] args) {
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              env.setParallelism(1);

              //准备几条河流
          DataStreamSource<String> ds1 = env.fromElements("a", "b", "C", "D", "E");
          DataStreamSource<String> ds2 = env.fromElements("hello", "world", "flink", "spark");
          DataStreamSource<Integer> ds3 = env.fromElements(1, 2, 3, 4, 5, 6, 7);

          //利用union合流
          //DataStream<String> unionDs = ds1.union(ds2, ds3.map(Objects::toString));
          //unionDs.print("UNION");

          //connect合流
          ConnectedStreams<String, Integer> integerConnectedStreams = ds1.connect(ds3);
          SingleOutputStreamOperator<String> connectDs = integerConnectedStreams.process(
                  new CoProcessFunction<String, Integer, String>() {
                      @Override
                      public void processElement1(String value, CoProcessFunction<String, Integer, String>.Context ctx, Collector<String> out) throws Exception {
                          //对第一条合流处理
                          out.collect(value.toUpperCase());//第一条合流转大写

                      }

                      @Override
                      public void processElement2(Integer value, CoProcessFunction<String, Integer, String>.Context ctx, Collector<String> out) throws Exception {
                        //对第二条合流处理
                          out.collect(value.toString());

                      }
                  }
          );
          connectDs.print("CONNECT");


          try {
                  env.execute();
              } catch (Exception e) {
                  throw new RuntimeException(e);
              }
          }
}
